package com.zengqingfa.examples.mybatisplus.handler;

import cn.hutool.core.collection.CollectionUtil;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.google.common.collect.Lists;
import com.zengqingfa.examples.mybatisplus.entity.User;
import com.zengqingfa.examples.mybatisplus.mapper.UserMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author zengqingfa
 * @className MultiThreadDataHandler
 * @description
 * @create 2022/9/8 20:33
 */
@Component
@Slf4j
public class MultiThreadDataHandler {

    public ExecutorService taskExecutor = new ThreadPoolExecutor(6, Integer.MAX_VALUE, 0, TimeUnit.MILLISECONDS, new SynchronousQueue<Runnable>());

    @Resource
    private UserMapper userMapper;

    /**
     * 多线程处理任务：使用for循环
     */
    public void handle() throws InterruptedException {
        // 每批次数量
        Integer pageSize = 20000;
        // 获取到要处理的数据的大小
        Integer count = userMapper.selectCount(null);
        int loopCount = (int) Math.ceil((float) count / pageSize);
        for (int k = 0; k < loopCount; k++) {
            Page<User> page = new Page<>(k, pageSize);
            IPage<User> userIPage = userMapper.selectUserPage1(page, null);
            List<User> records = userIPage.getRecords();
            List<List<User>> partition = Lists.partition(records, 2000);
            // 采用线程池的同步计数类
            CountDownLatch countDownLatch = new CountDownLatch(partition.size());
            partition.forEach(p -> {
                taskExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            List<User> handList = p;
                            //执行业务逻辑 TODO

                        } catch (Exception ex) {
                            log.error("getQwAllUserInfo error:", ex);
                        } finally {
                            //执行一个任务-1次
                            countDownLatch.countDown();
                        }
                    }
                });
            });
            //最后等所有线程数据处理完毕
            countDownLatch.await();
        }
    }

    /**
     * 多线程处理任务:使用while循环
     */
    public void handle2() throws InterruptedException {
        // 每批次数量
        Integer pageSize = 20000;
        Integer pageNum = 1;
        while (true) {
            Page<User> page = new Page<>(pageNum, pageSize);
            IPage<User> userIPage = userMapper.selectUserPage1(page, null);
            List<User> records = userIPage.getRecords();
            if (CollectionUtil.isEmpty(records)) {
                break;
            }
            List<List<User>> partition = Lists.partition(records, 2000);
            // 采用线程池的同步计数类
            CountDownLatch countDownLatch = new CountDownLatch(partition.size());
            partition.forEach(p -> {
                taskExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            List<User> handList = p;
                            //执行业务逻辑 TODO

                        } catch (Exception ex) {
                            log.error("getQwAllUserInfo error:", ex);
                        } finally {
                            //执行一个任务-1次
                            countDownLatch.countDown();
                        }
                    }
                });
            });
            //最后等所有线程数据处理完毕
            countDownLatch.await();
            pageNum++;
        }


    }
}